/*
 * Hyuga: A Density-Grid Stream Clustering Platform.
 * Copyright (C) 2014 PUC-Rio/Laboratory for Advanced Collaboration
 *
 * Hyuga is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * Hyuga is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with Hyuga.  If not, see <http://www.gnu.org/licenses/>.
 */
package br.pucrio.inf.lac.konoha.hyuga.listeners;

import org.apache.log4j.Logger;

import br.pucrio.inf.lac.konoha.hyuga.core.DDSLink;
import br.pucrio.inf.lac.konoha.hyuga.events.CEPEvent;

import com.espertech.esper.client.EventBean;
import com.espertech.esper.client.UpdateListener;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;

public class PublishOutputListener implements UpdateListener {
    
    /** DDSLink (Data Stream). */
    private DDSLink ddsLink;

    // JSON Mapper
    ObjectMapper mapper;

    /** LOGGER. */
    private static Logger LOGGER = Logger.getLogger(PublishOutputListener.class.getCanonicalName());
    
    /**
     * Default Constructor. Receive the DDSLink to send updates to other instances or management apps.
     * @param ddsLink - DDSLink
     */
    public PublishOutputListener(DDSLink ddsLink) {
        this.ddsLink = ddsLink;
        this.mapper  = new ObjectMapper();
    }

    @Override
    public void update(EventBean[] newEvents, EventBean[] oldEvents) {
        if (newEvents == null) { return; }

        CEPEvent cepEvent = (CEPEvent) newEvents[0].getUnderlying();
        
        try {
            ddsLink.publish(cepEvent.getType().toString(), mapper.writeValueAsBytes(cepEvent));
            LOGGER.debug("SEND [ " + cepEvent.getType().toString() + " ] " + cepEvent.toString());
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
    }

}
